package 操作zookeeper.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * author Veng Su
 * email  1344114844@qq.com
 * date   18-9-16 下午6:32
 */
public class CuratorDemo {
    public final static String CONNECTSTRING = "192.168.0.201:2181,192.168.0.203:2181,192.168.0.204:2181";

    public static CuratorFramework getInstance() {
        return CuratorFrameworkFactory
                .builder()
                .connectString(CONNECTSTRING)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 5))
                .build();
    }

//    测试创建

    @Test
    public void testCreate() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();
        curatorFramework.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath("/suveng/testcurator");

    }


    //    测试删除

    @Test
    public void testDelete() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();
        curatorFramework.delete()
                .deletingChildrenIfNeeded()
//                .withVersion()
                .forPath("/suveng/testcurator");

    }

    //测试getdata
    @Test
    public void testGet() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();
        Stat stat = new Stat();
        byte[] bytes = curatorFramework.getData()
                .storingStatIn(stat)
                .forPath("/suveng");
        System.out.println(new String(bytes));
        System.out.println("stat->" + stat);
    }

    //    测试setdata
    @Test
    public void testSet() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();
        curatorFramework.setData().forPath("/suveng", "knickknack".getBytes());
    }

    //测试异步
    @Test
    public void testasynchronous() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        CountDownLatch countDownLatch = new CountDownLatch(1);
        //创建临时节点
        curatorFramework.create().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println(Thread.currentThread().getName() + "这是异步的回调函数" + "   result code->" + curatorEvent.getResultCode() + "" +
                        "curator type" + curatorEvent.getType());
                countDownLatch.countDown();
            }
        }, executorService).forPath("/suveng/linshi");//不给其他线程默认是当前线程。
        countDownLatch.await();
        System.out.println(Thread.currentThread().getName() + "main线程");
        executorService.shutdown();


    }

    //事务
    @Test
    public void testTransaction() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();

        Collection<CuratorTransactionResult> curatorTransactions = curatorFramework.inTransaction().create().withMode(CreateMode.EPHEMERAL).forPath("/suveng/transaction")
                .and()
                .setData()
                .forPath("/suveng", "sdfsdfsd".getBytes()).and()
                .commit();
        for (CuratorTransactionResult result : curatorTransactions) {
            System.out.println(result.getForPath() + "->" + result.getType());
        }
    }

    //    测试watcher
//    patchcache 监视一个节点的子节点下的创建，删除，更新
//    nodecache 监视一个节点的创建，删除，更新
//    Treecache     patchcache + nodecache 合体 监视路径下的创建，删除，更新，并且缓存路径 所有子节点的数据
    @Test
    public void testWatcher() throws Exception {
        CuratorFramework curatorFramework = CuratorDemo.getInstance();
        curatorFramework.start();

//        NodeCache nodeCache=new NodeCache(curatorFramework,"/suveng",false);
//        nodeCache.start();
//
//        nodeCache.getListenable().addListener(()-> System.out.println("节点数据发生改变，改变后的数据"+new String(nodeCache.getCurrentData().getData())));
//        curatorFramework.setData().forPath("/suveng","sdfasfasfda".getBytes());
//        Thread.sleep(2000);


        PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, "/suveng", true);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        childrenCache.getListenable().addListener((curatorFramework1, pathChildrenCacheEvent) -> {
            switch (pathChildrenCacheEvent.getType()) {
                case CONNECTION_RECONNECTED:
                    System.out.println("reconnetc");
                    break;
                case CHILD_ADDED:
                    System.out.println("child add");
                    break;
                case CHILD_REMOVED:
                    System.out.println("child remove");
                    break;
                case CHILD_UPDATED:
                    System.out.println("child update");
                    break;
                default:
                    break;
            }
        });

//        curatorFramework.setData().forPath("/suveng","sdfsd".getBytes());

//        curatorFramework.delete().forPath("/suveng/test");
        curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/suveng/test");
        TimeUnit.SECONDS.sleep(1);
        curatorFramework.setData().forPath("/suveng/test","sfs".getBytes());
        TimeUnit.SECONDS.sleep(1);

    }

    @Test
    public void testInit() {
//        创建会话的两种方式
//        第一种 normal
        CuratorFramework curatorFramework = CuratorFrameworkFactory
                .newClient(CONNECTSTRING, 5000, 5000, new ExponentialBackoffRetry(1000, 3));
        curatorFramework.start();
//        第二种 fluent
        CuratorFramework curatorFramework2 = CuratorFrameworkFactory
                .builder()
                .connectString(CONNECTSTRING)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 5))
                .build();
        curatorFramework2.start();
        System.out.println("success ");

    }
}
